package drds.binlog.parse.inbound.mysql;

import drds.binlog.common.position.EntryPosition;
import drds.binlog.common.position.LogPosition;
import drds.binlog.parse.EventParser;
import drds.binlog.parse.exception.BinlogParseException;
import drds.binlog.parse.inbound.Connection;
import drds.binlog.parse.inbound.mysql.dbsync.LogEventConvert;
import drds.binlog.parse.inbound.mysql.dbsync.TableMetaDataCache;
import drds.binlog.parse.inbound.mysql.tsdb.DatabaseTableMetaData;
import drds.binlog.parse.index.LogPositionManager;
import drds.binlog.parse.support.AuthenticationInfo;
import org.apache.commons.lang.StringUtils;

import java.io.IOException;

/**
 * 基于本地binlog文件的复制
 */
public class LocalBinlogEventParser extends AbstractMysqlEventParser implements EventParser
{

    // 数据库信息
    protected AuthenticationInfo masterInfo;
    protected EntryPosition masterPosition;        // binlog信息
    protected MysqlConnection metaConnection;        // 查询meta信息的链接
    protected TableMetaDataCache tableMetaDataCache;        // 对应meta

    protected String directory;
    protected boolean needWait = false;
    protected int bufferSize = 16 * 1024;

    public LocalBinlogEventParser()
    {
        // this.runningInfo = new AuthenticationInfo();
    }

    @Override
    protected Connection buildErosaConnection()
    {
        return buildLocalBinLogConnection();
    }

    @Override
    protected void preDump(Connection connection)
    {
        metaConnection = buildMysqlConnection();
        try
        {
            metaConnection.connect();
        } catch (IOException e)
        {
            throw new BinlogParseException(e);
        }

        if (tableMetaTSDB != null && tableMetaTSDB instanceof DatabaseTableMetaData)
        {
            ((DatabaseTableMetaData) tableMetaTSDB).setConnection(metaConnection);
            ((DatabaseTableMetaData) tableMetaTSDB).setFilter(eventFilter);
            ((DatabaseTableMetaData) tableMetaTSDB).setBlackFilter(eventBlackFilter);
        }

        tableMetaDataCache = new TableMetaDataCache(metaConnection, tableMetaTSDB);
        ((LogEventConvert) binlogParser).setTableMetaDataCache(tableMetaDataCache);
    }

    @Override
    protected void afterDump(Connection connection)
    {
        if (metaConnection != null)
        {
            try
            {
                metaConnection.disconnect();
            } catch (IOException e)
            {
                logger.error("ERROR # disconnect metadata connection for address:{}", metaConnection.getConnector()
                        .getSocketAddress(), e);
            }
        }
    }

    public void start() throws BinlogParseException
    {
        if (runningInfo == null)
        { // 第一次链接主库
            runningInfo = masterInfo;
        }

        super.start();
    }

    @Override
    public void stop()
    {
        if (metaConnection != null)
        {
            try
            {
                metaConnection.disconnect();
            } catch (IOException e)
            {
                logger.error("ERROR # disconnect metadata connection for address:{}", metaConnection.getConnector()
                        .getSocketAddress(), e);
            }
        }

        if (tableMetaDataCache != null)
        {
            tableMetaDataCache.clearTableMeta();
        }

        super.stop();
    }

    private Connection buildLocalBinLogConnection()
    {
        LocalBinLogConnection localBinLogConnection = new LocalBinLogConnection();

        localBinLogConnection.setBufferSize(this.bufferSize);
        localBinLogConnection.setDirectory(this.directory);
        localBinLogConnection.setNeedWait(this.needWait);

        return localBinLogConnection;
    }

    private MysqlConnection buildMysqlConnection()
    {
        MysqlConnection mysqlConnection = new MysqlConnection(runningInfo.getInetSocketAddress(),
                runningInfo.getUsername(),
                runningInfo.getPassword(),
                connectionCharsetNumber,
                runningInfo.getDefaultDatabaseName());
        mysqlConnection.getConnector().setReceiveBufferSize(64 * 1024);
        mysqlConnection.getConnector().setSendBufferSize(64 * 1024);
        mysqlConnection.getConnector().setSoTimeout(30 * 1000);
        mysqlConnection.setCharset(connectionCharset);
        return mysqlConnection;
    }

    @Override
    protected EntryPosition findStartPosition(Connection connection)
    {
        // 处理逻辑
        // 1. 首先查询上一次解析成功的最后一条记录
        // 2. 存在最后一条记录，判断一下当前记录是否发生过主备切换
        // // a. 无机器切换，直接返回
        // // b. 存在机器切换，按最后一条记录的stamptime进行查找
        // 3. 不存在最后一条记录，则从默认的位置开始启动
        LogPosition logPosition = logPositionManager.getLatestIndexBy(destination);
        if (logPosition == null)
        {// 找不到历史成功记录
            EntryPosition entryPosition = masterPosition;

            // 判断一下是否需要按时间订阅
            if (StringUtils.isEmpty(entryPosition.getJournalName()))
            {
                // 如果没有指定binlogName，尝试按照timestamp进行查找
                if (entryPosition.getTimestamp() != null)
                {
                    return new EntryPosition(entryPosition.getTimestamp());
                }
            } else
            {
                if (entryPosition.getPosition() != null)
                {
                    // 如果指定binlogName + offest，直接返回
                    return entryPosition;
                } else
                {
                    return new EntryPosition(entryPosition.getTimestamp());
                }
            }
        } else
        {
            return logPosition.getEntryPosition();
        }

        return null;
    }

    // ========================= setter / getter =========================

    public void setLogPositionManager(LogPositionManager logPositionManager)
    {
        this.logPositionManager = logPositionManager;
    }

    public void setDirectory(String directory)
    {
        this.directory = directory;
    }

    public void setBufferSize(int bufferSize)
    {
        this.bufferSize = bufferSize;
    }

    public void setMasterPosition(EntryPosition masterPosition)
    {
        this.masterPosition = masterPosition;
    }

    public void setMasterInfo(AuthenticationInfo masterInfo)
    {
        this.masterInfo = masterInfo;
    }

    public boolean isNeedWait()
    {
        return needWait;
    }

    public void setNeedWait(boolean needWait)
    {
        this.needWait = needWait;
    }
}
